package cn.admin.configcenter.registry.service.client;

import cn.admin.configcenter.registry.common.cache.PatternCache;
import cn.admin.configcenter.registry.common.info.ClientConnection;
import cn.admin.configcenter.registry.common.info.GroupId;
import cn.admin.configcenter.registry.common.info.RevisedData;
import cn.admin.configcenter.registry.pojo.PurePublisherInfo;
import cn.admin.configcenter.registry.pojo.PureSubscriberInfo;
import cn.admin.configcenter.registry.service.DataService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.Future;
import java.util.regex.Pattern;

public class DefaultClientManager implements ClientManager {

    private static final Logger LOG = LoggerFactory.getLogger(DefaultClientManager.class);
    private final DataService dataService;

    public DefaultClientManager(DataService dataService) {
        this.dataService = dataService;
    }

    @Override
    public List<PurePublisherInfo> getPublisherInfo(String dataIdPattern,
                                                    String groupIdPattern, String ipPattern) throws Exception {
        Pattern dataIdPn = PatternCache.getPattern(dataIdPattern);
        Pattern groupIdPn = PatternCache.getPattern(groupIdPattern);
        Pattern ipPn = PatternCache.getPattern(ipPattern);
        final ClientFilter filter = new ClientFilter(dataIdPn,groupIdPn,ipPn);
        List<Future<List<PurePublisherInfo>>> futures = ClientConnectionWorkers.dispatchToAllWorkers(() -> {
            List<PurePublisherInfo> pubs = new LinkedList<>();
            List<ClientConnection> snapShots =
                    ClientConnectionWorker.getWorker().allConnectionSnapshot();
            for (ClientConnection client : snapShots) {
                for (Map.Entry<GroupId, RevisedData> entry : client.getPubData().entrySet()) {
                    GroupId group = entry.getKey();
                    String dataId = group.getDataId();
                    String groupId = group.getGroupId();
                    String hostId = client.originalHostId();
                    if (filter.accept(dataId,groupId,hostId)) {
                        PurePublisherInfo info = new PurePublisherInfo();
                        info.setDataId(dataId);
                        info.setHostId(hostId);
                        info.setGroups(Arrays.asList(groupId));
                        info.setPersistent(group.isPersist());
                        info.setClusterPublisher(!client.isNative());
                        info.setDatumId(hostId);
                        info.setClientId(client.getPublisherClientId(group));
                        info.setData(Arrays.asList(entry.getValue().getData()));
                        pubs.add(info);
                    }
                }
            }
            return pubs;
        });
        List<PurePublisherInfo> pubs = new LinkedList<>();
        for (Future<List<PurePublisherInfo>> future : futures) {
            try {
                pubs.addAll(future.get());
            } catch (Exception e) {
                LOG.error(e.toString(),e);
            }
        }
        return pubs;
    }

    @Override
    public List<PureSubscriberInfo> getSubscriberInfo(String dataIdPattern,
                                                      String groupIdPattern, String ipPattern) throws Exception {

        Pattern dataIdPn = PatternCache.getPattern(dataIdPattern);
        Pattern groupIdPn = PatternCache.getPattern(groupIdPattern);
        Pattern ipPn = PatternCache.getPattern(ipPattern);
        final ClientFilter filter = new ClientFilter(dataIdPn,groupIdPn,ipPn);
        List<Future<List<PureSubscriberInfo>>> futures = ClientConnectionWorkers.dispatchToAllWorkers(() -> {
            List<PureSubscriberInfo> subs = new LinkedList<>();
            List<ClientConnection> snapShots =
                    ClientConnectionWorker.getWorker().allConnectionSnapshot();
            for (ClientConnection client : snapShots) {
                Set<GroupId> groupIds = client.getSubscribers();
                for (GroupId group : groupIds) {
                    PureSubscriberInfo info = new PureSubscriberInfo();
                    info.setDataId(group.getDataId());
                    info.setGroups(Collections.singletonList(group.groupId));
                    info.setHostId(client.originalHostId());
                    info.setClientId(client.isSubscriber(group));
                    subs.add(info);
                }
            }
            return subs;
        });
        List<PureSubscriberInfo> subs = new LinkedList<>();
        for (Future<List<PureSubscriberInfo>> future : futures) {
            try {
                subs.addAll(future.get());
            } catch (Exception e) {
                LOG.error(e.toString(),e);
            }
        }
        subs.removeIf(pureSubscriberInfo -> !filter.accept(pureSubscriberInfo));
        return subs;
    }
}
